package com.gzsxy.rabbitmq.producer.config;

import cn.hutool.core.util.StrUtil;
import com.gzsxy.rabbitmq.producer.ConstantKey;
import com.gzsxy.rabbitmq.producer.config.redis.StringRedisTemplate;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;




/**
 * <p>
 * 定义生产者确认回调对象
 * 需要配置spring.rabbitmq.publisher-confirms=true  低版本
 * 或spring.rabbitmq.publisher-confirm-type=correlated 高版本
 * 每个RabbitTemplate只支持一个ConfirmCallback
 * 当消息没有到达Rabbitmq的交换机时触发该方法（当然到达了也会触发，）
 * </p>
 *
 * @Author: xiaolong
 * @date: 2023/2/13 16:31
 * @Version: 1.0
 */
@Configuration
@Slf4j
public class RabbitmqConfig {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private RabbitmqProperties config;


    /**
     *    解决mq消息丢失问题思路:
     *    把消息发送MQ一份，同时存储到数据库(Redis)一份。开启消息发送确定机制，然后进行消息的状态控制，发送中为0，发送失败为1。
     *    必须结合应答ACK(确认字符)来完成。对于那些如果没有发送成功的消息改redis消息状态为1，可以采用定时器进行轮询发送。
     * @params:
     * @return:
     * @date: 2023/2/15 10:38
     */
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //使用return-callback时必须设置mandatory为true才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数，或者在配置中设置mandatory-expression的值为true
        rabbitTemplate.setMandatory(true);
        // 如果消息没有到exchange,则confirm回调,ack=false; 如果消息到达exchange,则confirm回调,ack=true
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                //有些没有设置发送应答ack的，不需要走后续的逻辑
                if (correlationData == null) {
                    return;
                }
                String id = correlationData.getId();
                // 确认方法
                if (ack) {
                    //如果存在当前发送失败异常消息，则删除缓存
                    if (stringRedisTemplate.opsForHash().hasKey(ConstantKey.sendMsgStatus.FAIL.getValue(),id)){
                        stringRedisTemplate.opsForHash().delete(ConstantKey.sendMsgStatus.FAIL.getValue(),id);
                    }
                    // ack返回true说明mq接收到消息，删除redis中记录的消息
                    log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
                    stringRedisTemplate.opsForHash().delete(ConstantKey.sendMsgStatus.SEND.getValue(),id);
                } else {
                    // 如果为false，代表mq没有接收到消息(消息生产失败)
                    // 业务处理
                    //1.方式一：这里可以持久化业务消息体到数据库，然后定时去进行补偿处理或者重试等等
                    //使用这种方式获取消息，需要在发消息的时间进行封装correlationData对象correlationData.setReturnedMessage
                    //Message message=correlationData.getReturnedMessage();
                    //String msg = new String(message.getBody(), StandardCharsets.UTF_8);

                    //2.方式二：在这里如果生产失败，将redis消息修改状态1，通过定时任务去重新发送
                    //Object o = stringRedisTemplate.opsForValue().get(id); 该方式在消费者会导致该key未及时删除，导致缓存异常占用
                    Object o = stringRedisTemplate.opsForHash().get(ConstantKey.sendMsgStatus.SEND.getValue(), id);
                    stringRedisTemplate.opsForHash().put(ConstantKey.sendMsgStatus.FAIL.getValue(),id,o);
                    stringRedisTemplate.opsForHash().delete(ConstantKey.sendMsgStatus.SEND.getValue(),id);
                    log.info("消息发送失败:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
                }

            }
        });

        //如果exchange到queue成功,则不回调return;如果exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                //使用了x-delayed-message延迟插件，结果每次都强制触发returnedMessage回调方法？？？？
                //如果配置了发送回调ReturnCallback，插件延迟队列则会回调该方法，因为发送方确实没有投递到队列上，只是在交换器上暂存，等过期时间到了才会发往队列。
                //并非是BUG，而是有原因的，建议利用if去拦截这个异常，判断延迟队列交换机名称，然后break;
                if (StrUtil.equalsAny(exchange,config.getDelayedExchangeName())){
                    return;
                }


                log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message);
                byte[] body = message.getBody();
//                try {
//                    if (burnConfig.getTaskInfoRoutingkey().equals(routingKey)) {
//                        BurnTaskInfoDto burnTaskInfoDto = objectMapper.readValue(body, BurnTaskInfoDto.class);
//                        burnService.dealBurnTaskInfoDto(burnTaskInfoDto);
//                    } else if (burnConfig.getDelTaskReqRoutingkey().equals(routingKey)) {
//                        BurnCancelTaskDto burnCancelTaskDto = objectMapper.readValue(body, BurnCancelTaskDto.class);
//                        burnService.dealBurnCancelTaskDto(burnCancelTaskDto);
//                    }
//                } catch (IOException e) {
//                    log.error("处理消息丢失异常", e);
//                }
            }
        });
        return rabbitTemplate;
    }
}
